/*
 * SmartCam master processor library
 * Coprocessor message interpretation.
 */

#define _XOPEN_SOURCE 500

#include <log.h>
#include <scc.h>
#include <scu.h>
#include <settings.h>

#include "globals.h"
#include "management.h"
#include "dispatch.h"
#include "results.h"

scos_mutex_t scop_finished_mutex; 
scos_cond_t scop_finished_cond; 

static table_t deadlocks;

void scResResolveDeadlock(scs_target_operation_argument_t *stoa);
void *scResDeadlockThread(void *dummy);

void scResInit(void)
{
  lprintf(STREAM(scl.notice), "Initializing result gathering");

  scos_mutex_init(&scop_finished_mutex);
  scos_cond_init(&scop_finished_cond);
  
  abortOnError(STREAM(scl.err), tblcreate(&deadlocks, sizeof(scs_target_operation_argument_t)));
}

void scResCallback(struct scc_hdr_s *h)
{
  sc_op_t *o;
  sc_arg_t *a;
  sc_var_t *v;
  sc_buf_t *b;
  scc_msg_t *m = (scc_msg_t*)h;
  scs_operation_t *so;
  scs_target_operation_t *sto;
  scs_target_operation_argument_t *stoa;
  int id;

  scManAcquireLock();
  
  switch (m->header.type)
  {
    case SCC_MSGTYPE_OPERATIONCOMPLETED:
      /* Coprocessor signifies operation completion */
      abortOnNULL(STREAM(scl.err), tblget(&sc_ops, m->opid, &o));

      o->finished = 1;
      scManCheckDestroyOperation(o);

      scos_mutex_lock(&scop_finished_mutex);
      scos_cond_signal(&scop_finished_cond);
      scos_mutex_unlock(&scop_finished_mutex);
      break;
    case SCC_MSGTYPE_BUFFERCOMPLETED:
      /* oprocessor signifies buffer completion
       * (means: no more references) */
      abortOnNULL(STREAM(scl.err), tblget(&sc_bufs, m->argid, &b));
      b->destroyed = 1;
      b->file = *(int*)&m->data[sizeof(int)];
      b->stream->flowsize = *(int*)&m->data[0];
      
      lprintf(STREAM(scl.info), "Buffer %d (file %d, flow %d) completed",
              m->argid, b->file, b->stream->flowsize);

      scos_mutex_lock(&b->stream->lock);
      scos_cond_signal(&b->stream->condition);
      scos_mutex_unlock(&b->stream->lock);

      scManCheckDestroyBuffer(b);
      break;
    case SCC_MSGTYPE_BUFFERDEADLOCKED:
      /* Deadlock while writing buffer X */
      
      abortOnNULL(STREAM(scl.err), tblget(&sc_ops, m->opid, &o));
      abortOnNULL(STREAM(scl.err), tblget(&o->arguments, m->argid, &a));
      
      v = a->instance->variable;
      
      /* Add all arguments referencing this stream to deadlock list */
      do
      {
        abortOnNULL(STREAM(scl.err), tblget(&scs_operations, a->operation->type, &so));
        abortOnNULL(STREAM(scl.err), tblget(&so->targets,    a->operation->processor->id, &sto));
        abortOnNULL(STREAM(scl.err), tblget(&sto->arguments, a->id, &stoa));
        scResResolveDeadlock(stoa);
      } while (tblgetnextelement(&v->consumers, &id, &a));

      break;
    case SCC_MSGTYPE_OPERATIONRESULT:
      /* Coprocessor writes result variable */
      abortOnNULL(STREAM(scl.err), tblget(&sc_ops, m->opid, &o));
      abortOnNULL(STREAM(scl.err), tblget(&o->arguments, m->argid, &a));
      a->dispatched = 1;
      scResWriteResult(a->instance->variable, m->data);
      scManCheckDestroyArgument(a);
      break;
    default:
      lprintf(STREAM(scl.warning), "Received unrecognized message type %d from processor %d", m->header.type, m->header.source);
  }

  scManShedLock();  
}

void scResWriteResult(sc_var_t *v, void *data)
{
  int id;
  sc_instance_t *i;

  scos_mutex_lock(&v->lock);

  lprintf(STREAM(scl.info), "Writing data to variable %d type %d", v->id, v->type);

  abortOnNULL(STREAM(scl.err), v->data = malloc(v->size));
  memcpy(v->data, data, v->size);

  id = -1;
  while ((id = tblgetnexti(&v->instances, id)) >= 0)
  {
    abortOnNULL(STREAM(scl.err), tblget(&v->instances, id, &i));

    if (!i->finalized || i->blocked)
    {
      lprintf(STREAM(scl.crawl), "Writing %d bytes to %p", v->size, i->address);
      memcpy(i->address, data, v->size);
      _finalize(i->address);
    }
  }

  v->valid = 1;

  scDisVariable(v);

  scos_cond_signal(&v->condition);
  scos_mutex_unlock(&v->lock);
}

void scResResolveDeadlock(scs_target_operation_argument_t *stoa)
{
  static int first = 1;

  lprintf(STREAM(scl.notice), "Adding deadlock of %s operation type %d argument %d",
          stoa->operation_argument->operation->name,
          stoa->operation_argument->operation->id,
          stoa->operation_argument->id);

  tbladd(&deadlocks, stoa);
  
  if (first)
  {
    pthread_t thread;
    
    lprintf(STREAM(scl.notice), "Initializing deadlock gathering");
    
    pthread_create(&thread, NULL, scResDeadlockThread, NULL);

    first = 0;
  }
}

void *scResDeadlockThread(void *dummy)
{
  int id=-1;
  scs_target_operation_argument_t *stoa=NULL, *minstoa=NULL;
  int optype, argid;
  scs_operation_t *so;
  scs_target_operation_t *sto;

  // Gather all deadlocks
  sleep(5);
  
  while (tblgetnextelement(&deadlocks, &id, &stoa))
    if ((!minstoa || stoa->buffer < minstoa->buffer) &&
        stoa->operation_argument->operation->id > 15)
      minstoa = stoa;

  abortOnNULL(STREAM(scl.err), minstoa);

  optype = minstoa->operation_argument->operation->id;
  argid = minstoa->operation_argument->id;

  lprintf(STREAM(scl.notice), "Increasing operation %d argument %d buffer", optype, argid);
  
  scSetInit();
  scSetRead(settings.avfile);
  scSetIndex();
  
  abortOnNULL(STREAM(scl.err), tblget(&scs_operations, optype, &so));
  
  while (tblgetnextelement(&so->targets, &id, &sto))
    if (tblget(&sto->arguments, argid, &stoa))
    {
      lprintf(STREAM(scl.notice), "Increasing %s operation %s target argument %d buffer from %d to %d",
              so->name, sto->target->name, stoa->operation_argument->id,
              stoa->buffer, MAX(1, stoa->buffer*2));
    
      stoa->buffer = MAX(1, stoa->buffer*2);
    }

  scSetWrite(settings.avfile);
  
  exit(2);

  return NULL;
}
